package net.bwie.realtime.vehicle.dwd.utils;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.util.Collector;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

/**
 * 区域信息工具类：封装预加载和JSON处理逻辑
 * @author ASUS
 */
public class AreaInfoUtil {
    // JSON解析器（工具类内部使用）
    private static final ObjectMapper objectMapper = new ObjectMapper();
    // 区域信息缓存（工具类内部使用）
    private static final Map<String, String> areaInfoCache = new HashMap<>();

    /**
     * 预加载区域信息到缓存
     */
    public static void preloadAreaInfo() {
        // 数据库连接信息（请修改为你的实际配置）
        String url = "jdbc:postgresql://node103:5432/smart_traffic";
        String user = "postgres";
        String password = "123456";

        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            // 加载PostgreSQL驱动
            Class.forName("org.postgresql.Driver");
            conn = DriverManager.getConnection(url, user, password);
            stmt = conn.createStatement();
            rs = stmt.executeQuery("SELECT area_id, area_name FROM public.t_area_info");

            // 存入缓存
            while (rs.next()) {
                String areaId = rs.getString("area_id");
                String areaName = rs.getString("area_name");
                areaInfoCache.put(areaId, areaName);
            }

            System.out.println("预加载区域信息完成，共加载 " + areaInfoCache.size() + " 条数据");

        } catch (Exception e) {
            System.err.println("预加载失败: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 关闭资源
            try { if (rs != null) {
                rs.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
            try { if (stmt != null) {
                stmt.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
            try { if (conn != null) {
                conn.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
        }
    }

    /**
     * 处理JSON流数据，添加区域名称
     */
    public static DataStream<String> processJson(DataStream<String> logStream) {
        return logStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String jsonStr, Collector<String> out) throws Exception {
                try {
                    JsonNode jsonNode = objectMapper.readTree(jsonStr);
                    ObjectNode objectNode = (ObjectNode) jsonNode;

                    // 获取areaId并查询缓存
                    String areaId = jsonNode.get("areaId").asText();
                    String areaName = areaInfoCache.getOrDefault(areaId, "未知区域");

                    // 添加区域名称字段
                    objectNode.put("areaName", areaName);

                    // 输出处理后的JSON字符串
                    out.collect(objectNode.toString());
                } catch (Exception e) {
                    // 异常处理
                    e.printStackTrace();
                }
            }
        });

    }
}
